package cn.edu.gmall.realtime.utils;

import cn.edu.gmall.realtime.bean.GmallConfig;
import cn.edu.gmall.realtime.bean.TransientSink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class ClickhouseUtil {
    //获取针对 ClickHouse 的 JdbcSink
    public static <T> SinkFunction getJdbcSink(String sql, JdbcStatementBuilder<T> jdbcStatementBuilder) {
        SinkFunction<T> sink = JdbcSink.<T>sink(
                sql,
                jdbcStatementBuilder,
                new JdbcExecutionOptions.Builder().withBatchSize(2).build(),
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .build()
        );

        return sink;
    }

    public static <T> SinkFunction getJdbcSink(String sql) {
        JdbcStatementBuilder<T> jdbcStatementBuilder = new JdbcStatementBuilder<T>() {
            @Override
            public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                Field[] fields = t.getClass().getDeclaredFields();
                int skipOffset = 0;

                for (int i = 0; i < fields.length; i++){
                    Field field = fields[i];
                    //通过反射获得字段上的注解
                    TransientSink transientSink = field.getAnnotation(TransientSink.class);
                    if (transientSink != null) {
                        // 如果存在该注解
                        System.out.println("跳过字段:" + field.getName());
                        skipOffset++;
                        continue;
                    }
                    field.setAccessible(true);
                    try {
                        Object o = field.get(t);
                        preparedStatement.setObject(i + 1 - skipOffset, o);
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        SinkFunction<T> sink = JdbcSink.<T>sink(
                sql,
                jdbcStatementBuilder,
                new JdbcExecutionOptions.Builder().withBatchSize(2).build(),
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .build()
        );

        return sink;
    }
}
